package com.hujiang.basic.search.zookeeper.zkNotify;

import com.hujiang.basic.search.handler.HandlerService;
import com.hujiang.basic.search.util.IpUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import java.nio.charset.StandardCharsets;



@Slf4j
public class DistributedNotify {

    private String controlQueue;
    private CuratorFramework client;
    private String zkPath;
    private HandlerService handlerService;
    private static final int ZK_SLEEP_TIME_MS = 1000;
    private static final int ZK_MAX_RETRIES = 3;

    public void initCuratorClient(String zkAddr, String path, String queue, HandlerService service)  {
        zkPath = path;
        handlerService = service;
        controlQueue = queue;
        client = CuratorFrameworkFactory.newClient(
                zkAddr, new ExponentialBackoffRetry(ZK_SLEEP_TIME_MS, ZK_MAX_RETRIES));
        client.start();
        NodeCache cache = new NodeCache(client, path);
        try {
            cache.start();
            if (client.getZookeeperClient().getZooKeeper().exists(this.zkPath, false) == null) {
                client.getZookeeperClient().getZooKeeper().
                        create(this.zkPath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            log.error("DistributedNotify initCuratorClient error:",e);
        }
        addListener(cache);
    }

    private void addListener(final NodeCache cache) {
        NodeCacheListener listener = new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                if (cache.getCurrentData() != null && cache.getCurrentData().getData() != null) {
                    log.info("Node changed:{},value:{}",
                            cache.getCurrentData().getPath(), cache.getCurrentData().getData());
                    String data = new String(cache.getCurrentData().getData(), StandardCharsets.UTF_8);
                    if ("true".equalsIgnoreCase(data)
                            && handlerService.listeningQueues.contains(controlQueue)) {
                        handlerService.stopListen(controlQueue);
                        handlerService.pause(controlQueue);
                        handlerService.listeningQueues.remove(controlQueue);
                        log.info("{} stop listen to queue {}", IpUtils.getNodeAddr(), controlQueue);
                    }
                    else {
                        if (!"true".equalsIgnoreCase(data)
                                && !handlerService.listeningQueues.contains(controlQueue)) {
                            handlerService.restart(controlQueue);
                            handlerService.listen(controlQueue);
                            handlerService.listeningQueues.add(controlQueue);
                            log.info("{} start listen to queue {}", IpUtils.getNodeAddr(), controlQueue);
                        }
                    }
                }
            }
        };
        cache.getListenable().addListener(listener);
    }

    public void setValue(String isFullIndexing){
        try {
            client.setData().forPath(zkPath, isFullIndexing.getBytes(StandardCharsets.UTF_8));
        } catch (KeeperException.NoNodeException nodeEx) {
            log.info("distributedNotify set data NoNodeException ",nodeEx);
            try {
                client.create().creatingParentsIfNeeded()
                        .forPath(zkPath, isFullIndexing.getBytes(StandardCharsets.UTF_8));
            }catch(Exception ex){
                log.error("distributedNotify create zkPath Exception ",ex);
            }
        }catch(Exception e){
            log.error("distributedNotify set data Exception ",e);
        }
    }
}

